package com.xgst.bmg.sse.service.impl;

import com.xgst.bmg.sse.exception.SSEException;
import com.xgst.bmg.sse.service.SSEService;
import com.xgst.bmg.sse.session.SSESession;
import com.xgst.bmg.sse.task.SSEHeartBeatTask;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Service;
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;

import java.util.Date;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;

/**
 * @BelongsProject: screen-win
 * @BelongsPackage: com.xgst.bmg.sse.service
 * @Author: xgstjs-11
 * @CreateTime: 2021-05-18 14:10:07
 * @Description: SSE Service Impl
 */
@Service
public class SSEServiceImpl implements SSEService {
    private static final Logger logger = LoggerFactory.getLogger(com.xgst.bmg.sse.service.impl.SSEServiceImpl.class);

    /**
     * 发送心跳线程池
     */
    private static ScheduledExecutorService heartbeatExecutors = Executors.newScheduledThreadPool(5);

    /**
     * 新建连接
     *
     * @param clientId 客户端ID
     * @return
     */
    @Override
    public SseEmitter start(String clientId) {
        if(SSESession.exist(clientId)){
            //long型毫秒值超时时间，设置为0L为永不超时
            SseEmitter emitter = new SseEmitter(0L);
            logger.info("MSG: SseConnect | EmitterHash: {} | ID: {} | Date: {}", emitter.hashCode(), clientId, new Date());
            //设置心跳间隔时间
            final ScheduledFuture<?> future = heartbeatExecutors.scheduleAtFixedRate(new SSEHeartBeatTask(clientId), 0, 5, TimeUnit.SECONDS);
            //如果id不存在，则新建连接和对应事件
            SSESession.add(clientId, emitter, future);
            emitter.onCompletion(() -> {
                logger.info("MSG: SseConnectCompletion | EmitterHash: {} |ID: {} | Date: {}", emitter.hashCode(), clientId, new Date());
                SSESession.onCompletion(clientId);
            });
            emitter.onTimeout(() -> {
                logger.error("MSG: SseConnectTimeout | EmitterHash: {} |ID: {} | Date: {}", emitter.hashCode(), clientId, new Date());
                SSESession.onError(clientId, new SSEException("TimeOut(clientId: " + clientId + ")"));
            });
            emitter.onError(t -> {
                logger.error("MSG: SseConnectError | EmitterHash: {} |ID: {} | Date: {}", emitter.hashCode(), clientId, new Date());
                SSESession.onError(clientId, new SSEException("Error(clientId: " + clientId + ")"));
            });
            return emitter;
        }else{
            return SSESession.get(clientId);
        }
    }

    /**
     * 发送数据
     *
     * @param clientId 客户端ID
     * @return
     */
    @Override
    public int send(String clientId, String msg) {
        if (SSESession.send(clientId, msg)) {
            return 200;
        }
        return 201;
    }

    /**
     * 关闭连接
     *
     * @param clientId 客户端ID
     * @return
     */
    @Override
    public String close(String clientId) {
        logger.info("MSG: SseConnectClose | ID: {} | Date: {}", clientId, new Date());
        if (SSESession.del(clientId)) return "Succeed!";
        return "Error!";
    }
}
